package com.devops.admin.controller;

import cn.hutool.core.util.StrUtil;
import com.devops.admin.config.bean.K8SClientInform;
import com.devops.admin.service.DOSK8sResourceEventService;
import com.devops.admin.util.PodStatusUtil;
import com.devops.common.utils.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.*;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author hyb22
 * @version 1.0.0
 * @description TODO
 * @date 2022/11/1 14:46
 */
@ServerEndpoint("/pod/event/{releaseId}/{type}")
@Component
@Slf4j
public class EventWebSocketController {

    /**
     * 存放每个客户端对应的Session对象。
     */
    private static final Map<String,Session> socketMap = new ConcurrentHashMap<>();

    private PipedOutputStream out;

    private PipedInputStream output;

    @OnOpen
    public void onOpen(@PathParam("releaseId") Integer releaseId,
                       @PathParam("type") Integer type,
                       Session session) {
        try {
            // 通过session获取当前的请求的参数作为socketMap的key
            String key = getKey(session);
            // 如果key不为空则表示参数有值可
            if (StrUtil.isNotBlank(key)) {
                // 将本次请求的session放入到map中
                socketMap.put(key, session);
                // 发送消息
                sendMessageToUser(key,releaseId,type);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * WebSocket请求关闭
     */
    @OnClose
    public void onClose(Session session) {
        try {
            String key = getKey(session);
            if (key != null) {
                // 获取bean
                DOSK8sResourceEventService resourceEventService = getDosk8sResourceEventService();
                // 关闭inform的连接
                resourceEventService.closeInform(Integer.valueOf(key));
                socketMap.remove(key);
//                closeStream(out);
//                closeStream(output);
            }
            log.debug("有一连接关闭 ");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @OnError
    public void onError(Throwable thr) {
        log.error("websocket error." + thr);
    }

    @OnMessage
    public String sendMessage(String msg){
        return msg;
    }

    /**
     * 获取key
     * @param session
     * @return
     */
    public String getKey(Session session){
        String key = null;
        try {
            // 获取本次连接传递的releaseId的值
            key = String.valueOf(session.getRequestParameterMap().get("releaseId").get(0));
        } catch (Exception e) {
            log.error("根据session获取key失败：",e.getMessage());
            e.printStackTrace();
        }
        return key;
    }

    /**
     * 发送数据
     * @param key
     */
    public void sendMessageToUser(String key, Integer releaseId, Integer type){
        // 如果key为空则直接返回不开启连接
        if (key == null) {
            return;
        }
        // 获取本次连接的session
        Session session = socketMap.get(key);
        if (session==null) {
            return;
        }
        // 如果本次连接没有打开，则直接返回
        if (!session.isOpen()) {
            return;
        }
        try {
            out = new PipedOutputStream();
            // 设置输出管道的大小为4096字节，默认的1024字节无法满足需求
            output = new PipedInputStream(4096);
            try {
                output.connect(out);
            } catch (IOException e) {
                e.printStackTrace();
            }
            DOSK8sResourceEventService resourceEventService = getDosk8sResourceEventService();
            PodStatusUtil thread = new PodStatusUtil(output, session);
            thread.start();
            K8SClientInform inform = SpringUtil.getBean(K8SClientInform.class);
            inform.addPipedMap(releaseId,out,output);
            // 查询pod的状态
            resourceEventService.watchNamespaceEvent(releaseId,type,out);

        } catch (Exception e) {
            log.error("发送消息失败----->{}",e.getMessage());
            e.printStackTrace();
        }
    }

    private static DOSK8sResourceEventService getDosk8sResourceEventService() {
        // 获取bean
        return SpringUtil.getBean(DOSK8sResourceEventService.class);
    }

    public void closeStream(Closeable stream) {
        if (stream != null) {
            try {
                stream.close();
            } catch (IOException e) {
                log.error("close websocket error.:" + e);
            }
        }
    }

}
